package dmp.beans.sparkcore

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import dmp.utils.RptUtil
/**
  * author:CN.CDG
  * Date:2019/2/14
  * Time:13:13
  * 广告在每个地域的投放情况用sparkcore实现
  **/
object AreaAnalyseRptV2 {
  def main(args: Array[String]): Unit = {
    //判断参数
    if(args.length!=2){
      println(
        """
          |输入参数不合法
          |请输入输入路径：
          |inputPath
          |outputPath
        """.stripMargin)
      sys.exit()
    }
    val Array(inputPath,outputPath)=args
    val conf=new SparkConf()
      .setAppName("AreaAnalyseRptV2")
      .setMaster("local[*]")
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    val spark=SparkSession
      .builder()
      .config(conf)
      .getOrCreate()
    val dfData: DataFrame = spark.read.parquet(inputPath)
    val rddData: RDD[Row] = dfData.rdd
    rddData.map(t=>{
      //取出需要的字段
     val reqMode= t.getAs[Int]("requestmode")
      val proNode=t.getAs[Int]("processnode")
      val effTive=t.getAs[Int]("iseffective")
      val bill=t.getAs[Int]("isbilling")
      val bid=t.getAs[Int]("isbid")
      val orderId=t.getAs[Int]("adorderid")
      val win=t.getAs[Int]("iswin")
      val winPrice=t.getAs[Double]("winprice")
      val adPayMent=t.getAs[Double]("adpayment")

      val reqList=RptUtil.caculateReq(reqMode,proNode)
      val rtbList=RptUtil.caculateRtb(effTive,bill,bid,orderId,win,winPrice,adPayMent)
      val showClickList=RptUtil.caculateShow(reqMode,effTive)
      ((t.getAs[String]("provincename"),t.getAs[String]("cityname")),reqList++rtbList++showClickList)
    }).reduceByKey((List1,List2)=>{
      List1.zip(List2).map(t=>t._1+t._2)
    }).map(t=>{
      t._1._1+","+t._1._2+","+t._2.mkString(",")  //mkString()将List集合转换成字符串
    })
      .saveAsTextFile(outputPath)
  }
}
